package com.lhd.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lhd.bean.PageSourceMetric;
import com.lhd.utils.MyClickHouseUtil;
import com.lhd.bean.PageSourceMetric;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

public class PageViewStatProcess {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));

        String pageTopic = "dwd_traffic_page_log";
        String orderTopic = "dwd_trade_order_detail";
        String consumerGroup = "page_source_metric_group";

        Properties kafkaProps = new Properties();
        kafkaProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        kafkaProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
        kafkaProps.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        kafkaProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        kafkaProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        DataStream<String> pageLogStream = env.addSource(
                new FlinkKafkaConsumer<>(pageTopic, new SimpleStringSchema(), kafkaProps)
        ).name("page-log-source");

        DataStream<String> orderDetailStream = env.addSource(
                new FlinkKafkaConsumer<>(orderTopic, new SimpleStringSchema(), kafkaProps)
        ).name("order-detail-source");

        // 3. 处理页面日志
        SingleOutputStreamOperator<PageVisitData> pageVisitStream = pageLogStream
                .map(jsonStr -> {
                    try {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject common = jsonObj.getJSONObject("common");
                        JSONObject page = jsonObj.getJSONObject("page");

                        if (common == null || page == null) return null;

                        String uid = common.getString("uid");
                        if (uid == null || uid.trim().isEmpty()) return null;

                        String sourceType = normalizeSourceType(page.getString("source_type"));
                        String shopId = common.getString("shop_id");
                        Long ts = jsonObj.getLong("ts");
                        String curDate = new SimpleDateFormat("yyyy-MM-dd").format(new Date(ts));

                        return new PageVisitData(uid, shopId, "", sourceType, ts, curDate);
                    } catch (Exception e) {
                        System.err.println("解析页面日志失败: " + jsonStr);
                        return null;
                    }
                })
                .filter(info -> info != null)
                .name("page-log-parse");

        // 4. 处理订单明细（关键：统一 sourceType）
        SingleOutputStreamOperator<OrderData> orderDataStream = orderDetailStream
                .map(jsonStr -> {
                    try {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        String userId = jsonObj.getString("user_id");
                        if (userId == null || userId.trim().isEmpty()) return null;

                        String sourceTypeName = jsonObj.getString("source_type_name");
                        String normalizedSourceType = normalizeSourceType(sourceTypeName); // 统一命名

                        String orderId = jsonObj.getString("order_id");
                        String shopId = "shop_" + (orderId != null && orderId.length() >= 2 ? orderId.substring(0, 2) : "00");
                        String createTime = jsonObj.getString("create_time");
                        String curDate = (createTime != null && createTime.length() >= 10) ? createTime.substring(0, 10) :
                                new SimpleDateFormat("yyyy-MM-dd").format(new Date());

                        return new OrderData(userId, shopId, orderId, normalizedSourceType, System.currentTimeMillis(), curDate);
                    } catch (Exception e) {
                        System.err.println("解析订单数据失败: " + jsonStr);
                        return null;
                    }
                })
                .filter(info -> info != null)
                .name("order-data-parse");

        // 5. 转换为统一事件
        SingleOutputStreamOperator<MetricEvent> pageMetricStream = pageVisitStream
                .map(data -> new MetricEvent("page_visit", data.getUid(), data.getSourceType(), data.getShopId(), data.getCurDate()));

        SingleOutputStreamOperator<MetricEvent> orderMetricStream = orderDataStream
                .map(data -> new MetricEvent("order", data.getUserId(), data.getSourceTypeName(), data.getShopId(), data.getCurDate()));

        // 6. 合并流
        DataStream<MetricEvent> mergedStream = pageMetricStream.union(orderMetricStream)
                .filter(event -> event != null &&
                        event.getUserId() != null && !event.getUserId().trim().isEmpty() &&
                        event.getSourceType() != null && !event.getSourceType().trim().isEmpty());

        // 7. 聚合（每30秒）
        SingleOutputStreamOperator<PageSourceMetric> resultStream = mergedStream
                .keyBy(event -> event.getSourceType() + ":" + event.getStatDate())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
                .aggregate(new SourceMetricAggregator());

        resultStream.print("✅ Page Source Metric Result:");

        // 8. 写入 ClickHouse（建议：先注释此行测试，避免 Sink 阻塞）

        String clickhouseSql = "INSERT INTO gmall.dws_page_source_metric (source_type, visitor_count, buyer_count, conversion_rate, stat_date, etl_time) VALUES (?, ?, ?, ?, ?, ?)";
        resultStream.addSink(MyClickHouseUtil.getSinkFunction(clickhouseSql))
                .name("clickhouse-sink");


        env.execute("Page Source Metric Analysis Job");
    }

    // === 关键：统一来源类型命名 ===
    private static String normalizeSourceType(String sourceType) {
        if (sourceType == null || sourceType.trim().isEmpty()) {
            return "direct";
        }
        String s = sourceType.trim().toLowerCase();
        switch (s) {
            case "search":
            case "搜索":
                return "search";
            case "recommend":
            case "首页推荐":
            case "recommendation":
                return "recommend";
            case "notification":
            case "通知":
                return "notification";
            case "direct":
            case "直接访问":
                return "direct";
            default:
                return "other";
        }
    }

    // ========== 内部类 ==========
    public static class PageVisitData {
        private String uid, shopId, pageId, sourceType, curDate;
        private Long timestamp;

        public PageVisitData(String uid, String shopId, String pageId, String sourceType, Long timestamp, String curDate) {
            this.uid = uid; this.shopId = shopId; this.pageId = pageId;
            this.sourceType = sourceType; this.timestamp = timestamp; this.curDate = curDate;
        }

        public String getUid() { return uid; }
        public String getSourceType() { return sourceType; }
        public String getCurDate() { return curDate; }
        public String getShopId() { return shopId; }
    }

    public static class OrderData {
        private String userId, shopId, orderId, sourceTypeName, curDate;
        private Long opTs;

        public OrderData(String userId, String shopId, String orderId, String sourceTypeName, Long opTs, String curDate) {
            this.userId = userId; this.shopId = shopId; this.orderId = orderId;
            this.sourceTypeName = sourceTypeName; this.opTs = opTs; this.curDate = curDate;
        }

        public String getUserId() { return userId; }
        public String getSourceTypeName() { return sourceTypeName; }
        public String getCurDate() { return curDate; }
        public String getShopId() { return shopId; }
    }

    public static class MetricEvent {
        private String eventType, userId, sourceType, shopId, statDate;

        public MetricEvent(String eventType, String userId, String sourceType, String shopId, String statDate) {
            this.eventType = eventType; this.userId = userId; this.sourceType = sourceType;
            this.shopId = shopId; this.statDate = statDate;
        }

        public String getEventType() { return eventType; }
        public String getUserId() { return userId; }
        public String getSourceType() { return sourceType; }
        public String getStatDate() { return statDate; }
    }

    public static class MetricAggState {
        private Set<String> uniqueVisitors = new HashSet<>();
        private Set<String> uniqueBuyers = new HashSet<>();
        private String sourceType;
        private String statDate;

        public void addVisitor(String userId) {
            if (userId != null) uniqueVisitors.add(userId.trim());
        }

        public void addBuyer(String userId) {
            if (userId != null) uniqueBuyers.add(userId.trim());
        }

        public int getVisitorCount() { return uniqueVisitors.size(); }
        public int getBuyerCount() { return uniqueBuyers.size(); }

        public String getSourceType() { return sourceType; }
        public void setSourceType(String sourceType) { this.sourceType = sourceType; }
        public String getStatDate() { return statDate; }
        public void setStatDate(String statDate) { this.statDate = statDate; }
    }

    // ========== 聚合函数：确保 visitor 和 buyer 都能加入 ==========
    public static class SourceMetricAggregator implements AggregateFunction<MetricEvent, MetricAggState, PageSourceMetric> {
        @Override
        public MetricAggState createAccumulator() {
            return new MetricAggState();
        }

        @Override
        public MetricAggState add(MetricEvent event, MetricAggState acc) {
            // ✅ 关键：直接设置，不再跳过 "unknown"
            if (acc.getSourceType() == null) {
                acc.setSourceType(event.getSourceType());
            }
            if (acc.getStatDate() == null) {
                acc.setStatDate(event.getStatDate());
            }

            if ("page_visit".equals(event.getEventType())) {
                acc.addVisitor(event.getUserId());
            } else if ("order".equals(event.getEventType())) {
                acc.addBuyer(event.getUserId());
            }
            return acc;
        }

        @Override
        public PageSourceMetric getResult(MetricAggState acc) {
            int visitors = acc.getVisitorCount();
            int buyers = acc.getBuyerCount();
            double rate = visitors > 0 ? (double) buyers / visitors : 0.0;

            PageSourceMetric metric = new PageSourceMetric();
            metric.setSourceType(acc.getSourceType() != null ? acc.getSourceType() : "unknown");
            metric.setVisitorCount(visitors);
            metric.setBuyerCount(buyers);
            metric.setConversionRate(rate);
            metric.setStatDate(acc.getStatDate() != null ? acc.getStatDate() : new SimpleDateFormat("yyyy-MM-dd").format(new Date()));
            metric.setEtlTime(System.currentTimeMillis());

            return metric;
        }

        @Override
        public MetricAggState merge(MetricAggState a, MetricAggState b) {
            MetricAggState m = new MetricAggState();
            m.setSourceType(a.getSourceType() != null ? a.getSourceType() : b.getSourceType());
            m.setStatDate(a.getStatDate() != null ? a.getStatDate() : b.getStatDate());
            m.uniqueVisitors.addAll(a.uniqueVisitors);
            m.uniqueVisitors.addAll(b.uniqueVisitors);
            m.uniqueBuyers.addAll(a.uniqueBuyers);
            m.uniqueBuyers.addAll(b.uniqueBuyers);
            return m;
        }
    }
}